package com.bwstudio.demo.consumer.message;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.bwstudio.demo.consumer.model.AccountChangeEvent;
import com.bwstudio.demo.consumer.service.AccountInfoService;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
@RocketMQMessageListener(topic = "topic_txmsg", 
	selectorExpression = "v0",
	consumerGroup = "${spring.application.name}-txmsg-bank2-v0")
public class TxmsgConsumer implements RocketMQListener<String> {

	private abstract class TxMsgProcessTemplate {
		public final void doProcess(String jsonStr) {
			log.info("开始消费消息:{}", jsonStr);
			try {
					JSONObject json = doParseAndValidate(jsonStr);
					operateDB(json);
			} catch (Exception e) {
					log.error("topic_txmsg 消费失败", e);
					throw new RuntimeException(e.getMessage(), e);
			}
		}
		protected abstract JSONObject doParseAndValidate(String jsonStr) throws RuntimeException;
		protected abstract void operateDB(JSONObject jsonObject);
	}
	
    @Autowired
    AccountInfoService accountInfoService;

    //接收消息
    @Override
    public void onMessage(String message) {
        new TxMsgProcessTemplate() {
        	@Override
			protected JSONObject doParseAndValidate(String jsonStr) throws RuntimeException {
        		return  JSONObject.parseObject(jsonStr);
			}
			@Override
			protected void operateDB(JSONObject jsonObject) {
		        String accountChangeString = jsonObject.getString("accountChange");
		        //转成AccountChangeEvent
		        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
		        //设置账号为李四的
		        accountChangeEvent.setAccountNo("2");
		        //更新本地账户，增加金额
		        accountInfoService.addAccountInfoBalance(accountChangeEvent);
			}
        	
        }.doProcess(message);

    }
}
